package net.jacqeush.mini.rpc.client;

import com.aliyun.opensearch.sdk.dependencies.com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import net.jacqeush.mini.rpc.cache.ConnectCache;
import net.jacqeush.mini.rpc.client.listener.ChannelCloseListener;
import net.jacqeush.mini.rpc.client.listener.ChannelSimpleListener;
import net.jacqeush.mini.rpc.handler.AvailableAndWaitHandler;
import net.jacqeush.mini.rpc.handler.RpcClientHandler;
import net.jacqeush.mini.rpc.handler.RpcClientInitializer;
import org.apache.commons.collections4.CollectionUtils;

import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.sql.Connection;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @author Jacquesh
 * @version 1.0
 * @date 2020/10/15 17:15
 */
@Slf4j
public class RpcConnectManager {
    private static final RpcConnectManager RPC_CLIENT_MANAGER = new RpcConnectManager();
    private AtomicInteger atomic = new AtomicInteger(0);
    /**
     * 提交链接的线程池
     */
    ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("socket-connect-pool-%d").build();
    ExecutorService singleThreadPool = new ThreadPoolExecutor(16, 16,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>(Integer.MAX_VALUE),
            namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
    private final EventLoopGroup loopGroup = new NioEventLoopGroup(4);
    private volatile boolean isRunning = true;

    /**
     * 一个连接的地址，对应一个实际的业务处理器(client)
     */

    private RpcConnectManager() {
    }

    public static RpcConnectManager getInstance() {
        return RPC_CLIENT_MANAGER;
    }

    /**
     * $connect 链接
     *
     * @param serverAddress 链接地址
     */
    public void connect(final String serverAddress) {
        List<String> connectAddressList = Arrays.asList(serverAddress.split(","));
         updateAndConnected(connectAddressList);
    }

    /**
     * $updateAndConnected 更新缓存并异步链接
     *
     * @param connectAddressList 链接地址
     */
    public void updateAndConnected(List<String> connectAddressList) {
        if (CollectionUtils.isEmpty(connectAddressList)) {
            ConnectCache.getInstance().clearConnected();
            throw new RuntimeException("更新缓存并异步链接出错");
        } else {
            AtomicReference<HashSet<InetSocketAddress>> socketAddress = new AtomicReference<>(new HashSet<>());
            connectAddressList.forEach((String address) -> {
                String[] array = address.split(":");
                String host = array[0];
                int port = Integer.parseInt(array[1]);
                final InetSocketAddress inetSocketAddress = new InetSocketAddress(host, port);
                socketAddress.get().add(inetSocketAddress);
                if (!ConnectCache.getInstance().CONNECTED_HANDLER_MAP.containsKey(inetSocketAddress)) {
                    connectAsync(inetSocketAddress);
                }
            });
            for (int i = 0; i < ConnectCache.getInstance().CONNECTED_HANDLER_LIST.size(); i++) {
                RpcClientHandler rpcClientHandler = ConnectCache.getInstance().CONNECTED_HANDLER_LIST.get(i);
                SocketAddress address = rpcClientHandler.getNetChannel().remoteAddress();
                if (!socketAddress.get().contains(address)) {
                    log.info(" remove invalid server node " + address);
                    RpcClientHandler handler = ConnectCache.getInstance().CONNECTED_HANDLER_MAP.get(address);
                    if (handler != null) {
                        handler.close();
                        ConnectCache.getInstance().CONNECTED_HANDLER_MAP.remove(address);
                    }
                    ConnectCache.getInstance().CONNECTED_HANDLER_LIST.remove(rpcClientHandler);
                }
            }
        }
    }

    public void connectAsync(InetSocketAddress inetSocketAddress) {
        singleThreadPool.submit(() -> {
            Bootstrap bootstrap = new Bootstrap();
            final ChannelFuture channelFuture = bootstrap
                    .group(loopGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new RpcClientInitializer())
                    .connect(inetSocketAddress);
            //添加监听
            channelFuture.channel().closeFuture().addListener(new ChannelCloseListener());
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {

                        if (future.isSuccess()){
                            log.info("channel is succeed connect ,the channel net address is:"+future.channel().remoteAddress());
                            RpcClientHandler handler = future.channel().pipeline().get(RpcClientHandler.class);
                                ConnectCache.getInstance().cache(handler);
                                AvailableAndWaitHandler.available();
                        }
                    }

            });
        });
    }


    public RpcClientHandler chooseHandler() throws InterruptedException {
        CopyOnWriteArrayList<RpcClientHandler> handlers = ConnectCache.getInstance().getClone();

        int size = handlers.size();

        while (isRunning && size <= 0) {
            try {
                boolean available = AvailableAndWaitHandler.waiting();
                if (available) {
                    handlers = ConnectCache.getInstance().getClone();
                    ;
                    size = handlers.size();
                }
            } catch (InterruptedException e) {
                log.error("waiting for available node is interrupted !");
                throw new RuntimeException("no connect any servers!", e);
            }
        }
        if (!isRunning) {
            return null;
        }
        // 	最终使用取模方式取得其中一个业务处理器进行实际的业务处理
        return handlers.get(((atomic.getAndAdd(1) + size) % size));
    }

    /**
     * $ close 关闭链接
     */
    public void close() {
        isRunning = false;
        CopyOnWriteArrayList<RpcClientHandler> handlers = ConnectCache.getInstance().getClone();
        handlers.forEach(RpcClientHandler::close);
        AvailableAndWaitHandler.available();
        loopGroup.shutdownGracefully();
        singleThreadPool.shutdown();
    }

    /**
     * $ 重连
     * @param handler handler
     * @param address address地址
     */
    public void reconnect(final RpcClientHandler handler, final SocketAddress address) {
        if (handler != null) {
            handler.close();
            ConnectCache.getInstance().remove(handler);
        }
        connectAsync((InetSocketAddress) address);
    }

}
